package bigdata.hourglass.utils;


import bigdata.hourglass.common.domain.Bean;
import bigdata.hourglass.common.utils.ZkUtils;
import com.typesafe.config.Config;
import com.typesafe.config.ConfigFactory;
import org.I0Itec.zkclient.ZkClient;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import scala.collection.Iterator;
import scala.collection.Seq;

import java.util.ArrayList;
import java.util.List;

/**
 * @author leo.jie (weixiao.me@aliyun.com)
 * @version 1.0
 * @organization CC
 * @website https://www.jlpyyf.com
 * @date 2019/12/11 15:03
 * @since 1.0
 */
public class AkkaRpcProviderFactory {
    private static final Logger logger = LoggerFactory.getLogger(AkkaRpcProviderFactory.class);

    private String ip;
    private int port;
    private String logLevel;

    public AkkaRpcProviderFactory() {

    }

    public void initConfig(String ip, int port, String logLevel) {
        this.ip = ip;
        this.port = port;
        this.logLevel = logLevel;
        if (this.ip == null) {
            this.ip = IpUtil.getIp();
        }
        if (this.port <= 0) {
            this.port = 2550;
        }
        if (this.logLevel == null) {
            this.logLevel = "info";
        }
        if (NetUtil.isPortUsed(this.port)) {
            this.port = NetUtil.findAvailablePort(Integer.parseInt(HourglassSystemConstant.DEFAULT_AKKA_TRANSPORT));
        }
    }

    public void initConfig(int port) {
        this.initConfig(null, port, null);
    }

    public void initConfig(int port, String logLevel) {
        this.initConfig(null, port, logLevel);
    }

    public Config getAkkaConfig(ZkClient zkClient) {
        Seq<Bean.HourglassEngine> allEngineInCluster = ZkUtils.getAllEngineInCluster(zkClient);
        Iterator<Bean.HourglassEngine> iterator = allEngineInCluster.iterator();
        List<Integer> engineIds = new ArrayList<>();
        List<Integer> ports = new ArrayList<>();

        while (iterator.hasNext()) {
            Bean.HourglassEngine engine = iterator.next();
            engineIds.add(engine.engineId());
            if (engine.engineInfo().contains(this.ip)) {
                ports.add(Integer.valueOf(engine.engineInfo().split(":")[1]));
            }
        }
        if (ports.size() != 0) {
            while (ports.contains(this.port)) {
                this.port += 1;
            }
        }
        this.port = NetUtil.findAvailablePort(this.port);
        int id = 1;
        if (engineIds.size() != 0) {
            while (engineIds.contains(id)) {
                id += 1;
            }
        }
        ZkUtils.registerEngineInZk(zkClient, id, this.ip, this.port, 6000, -1);
        return createAkkaConfig(this.ip, this.port, this.logLevel);
    }


    public String getIp() {
        return ip;
    }

    public int getPort() {
        return port;
    }

    private Config createAkkaConfig(String ip, int port, String logLevel) {
        return ConfigFactory.parseString("akka.remote.netty.tcp.port=" + port)
                .withFallback(ConfigFactory.parseString("akka.loglevel=" + logLevel))
                .withFallback(ConfigFactory.parseString("akka.remote.netty.tcp.hostname=" + ip))
                .withFallback(ConfigFactory.parseString("akka.actor.provider=akka.remote.RemoteActorRefProvider"))
                .withFallback(ConfigFactory.parseString("akka.actor.warn-about-java-serializer-usage=false"))
                .withFallback(ConfigFactory.load());
    }
}
